package cn.kgc.gmall.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * 查询维度信息工具类
 */
public class DimUtil {
    /**
     * 重构dimInfo方法 修改传入的参数
     * @return
     */
    public static JSONObject getDimInfo(String tableName,String key){

        return getDimInfo(tableName,Tuple2.of("id",key));
    }
    /**
     * 增加旁路缓存版本
     * 1.先去redis中查询  如果redis中有维度信息直接获取
     *      维度信息以什么形式存入redis中:k: String dim:表名:id_id  v:  String:{name:zhangsan,age:10}
     * 2.如果redis中没有维度信息 去hbase中查询 并且在redis中缓存一份维度信息
     * @param tableName
     * @param colNameAndValue
     * @return JSONObject{name:zhangsan,age:10}
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String,String>... colNameAndValue){
        // 拼接redis_key
        StringBuilder redisKey = new StringBuilder();
        redisKey.append("dim:").append(tableName.toLowerCase()).append(":");
        // 拼接sql: select * from user_info where user_id = 1 and user_name = zhangsan and ...
        StringBuilder sql = new StringBuilder();
        sql.append(" select * from  ").append(tableName).append(" where ");
        // 拼接字段
        for (int i = 0; i < colNameAndValue.length; i++) {
            // 获取元组中条件名
            String fieldName = colNameAndValue[i].f0;
            String fieldValue = colNameAndValue[i].f1;
            if (i > 0){
                sql.append(" and ");
                redisKey.append("_");
            }
            sql.append(fieldName).append("='").append(fieldValue).append("'");
            redisKey.append(fieldValue);
        }
        //System.out.println("执行的sql语句是:"+sql);
        //System.out.println("拼接好的rediskey是:"+redisKey);

        // 首先去redis中查询
        Jedis jedis = null;
        String dimJson = null;
        JSONObject dimInfo = null;
        // 获取redis连接对象
        try {
            jedis = RedisUtil.getJedisPool();
            dimJson = jedis.get(redisKey.toString());

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        // 如果redis中没有查询到
        if (dimJson != null){
            // 将查到的json字符串转化为json对象进行返回
             dimInfo = JSON.parseObject(dimJson);
        }else{
            // 如果在缓存中没有查到就去hbase中查询
            // 执行sql语句
            List<JSONObject> jsonObjects = PhoenixUtil.queryList(sql.toString(), JSONObject.class);
            if (jsonObjects.size() > 0){
                // 维度信息只有一条
                dimInfo = jsonObjects.get(0);
                // 将hbase中查询到的json对象存入redis中进行缓存 缓存时间为1天
                jedis.setex(redisKey.toString(),60 * 60 * 24 ,dimInfo.toJSONString());

            }else{
                System.out.println("没有查到维度信息");
            }

        }
        jedis.close();
        return dimInfo;
    }

    /**
     * 如果hbase中的数据发生了修改和删除redis中缓存数据也要进行删除
     */
    public  static void deleteCached(String tableName,String ... fields){
        // 拼接redis_key
        StringBuilder redisKey = new StringBuilder();
        redisKey.append("dim:").append(tableName.toLowerCase()).append(":");
        for (int i = 0; i < fields.length; i++) {
            if (i > 0){
                redisKey.append("_");
            }
            redisKey.append(fields[i]);
        }
        // 清除redis中对应的缓存数据
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedisPool();
            jedis.del(redisKey.toString());
            jedis.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    // 表名  名 = 值
    // 假如: 查询订单维度信息 订单id:37006  用户id: 1  假如有联合主键

    // map:key user_id value 1  Map需要使用entrySet遍历结果  2元组更加方便
    // 最后查到的维度信息封装到orderWide里面 所以返回值使用jsonObject

    /**
     * 无缓存版本
     *
     */
    /*public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String,String> ... colNameAndValue){
        // 拼接sql: select * from user_info where user_id = 1 and user_name = zhangsan and ...
        StringBuilder sql = new StringBuilder();
        sql.append(" select * from  ");
        sql.append(tableName);
        sql.append(" where ");
        // 拼接字段
        for (int i = 0; i < colNameAndValue.length; i++) {
            // 获取元组中条件名
            String fieldName = colNameAndValue[i].f0;
            String fieldValue = colNameAndValue[i].f1;
            if (i > 0){
                sql.append(" and ");
            }
            sql.append(fieldName).append("='").append(fieldValue).append("'");
        }
        System.out.println("执行的sql语句是:"+sql);
        // 执行sql语句
        List<JSONObject> jsonObjects = PhoenixUtil.queryList(sql.toString(), JSONObject.class);
        JSONObject jsonObject = null;
        if (jsonObjects.size() > 0){
            // 维度信息只有一条
            jsonObject = jsonObjects.get(0);
        }else{
            System.out.println("没有查到维度信息");
        }
        return jsonObject;
    }*/

    /*public static void main(String[] args) {
        JSONObject dimInfoNoCache = getDimInfo("DIM_USER_INFO", Tuple2.of("ID", "1"));
        System.out.println(dimInfoNoCache);
    }*/
}
